package com.dcloud.mq.consumer;

import com.dcloud.mq.config.ConsumerConfig;
import com.dcloud.mq.config.RocketConfig;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

public class SimpleConsumer {
    private DefaultMQPushConsumer consumer;
    //rocketMq配置
    private RocketConfig config;
    //消费者配置
    private ConsumerConfig consumerConfig;
    //消息监听器
    private MessageListenerConcurrently messageListener;

    public SimpleConsumer(RocketConfig config, ConsumerConfig consumerConfig, MessageListenerConcurrently messageListener){
        this.messageListener = messageListener;
        this.config = config;
        this.consumerConfig = consumerConfig;
    }

    public void init() throws Exception{
        consumer = new DefaultMQPushConsumer(consumerConfig.getConsumerGroup());
        //服务地址
        consumer.setNamesrvAddr(config.getAddress());
        //实例名称
        if(consumerConfig.getInstanceName() != null) {
            consumer.setInstanceName(consumerConfig.getInstanceName());
        }
        //订阅主题
        consumer.subscribe("task_1", "*");
        //消息模式
        consumer.setMessageModel(consumerConfig.getMessageModel());
        //从哪开始消费
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        //监听器
        //consumer.registerMessageListener(messageListener);
        consumer.registerMessageListener(new MessageListenerConcurrently() {

            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                            ConsumeConcurrentlyContext context) {
                System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
    }

    public void destroy(){
        consumer.shutdown();
    }

    public DefaultMQPushConsumer getConsumer() {
        return consumer;
    }
}
